package cn.bby.query.server.application.manager.impl;

import cn.bby.jtk.base.exception.ServiceException;
import cn.bby.query.base.api.req.PushInterfaceInfoReq;
import cn.bby.query.base.constant.AdvQueryErrorCode;
import cn.bby.query.base.model.dto.InterfaceMetaInfo;
import cn.bby.query.server.application.manager.InterfaceInfoManager;
import cn.bby.query.server.domain.agg.metadata.entity.InterfaceInfo;
import cn.bby.query.server.domain.agg.metadata.repository.InterfaceInfoRepository;
import cn.bby.query.server.domain.service.fetcher.DataFetcher;
import cn.bby.query.server.domain.service.manager.DataStorageManager;
import cn.bby.query.server.domain.share.constant.InterfaceStatus;
import cn.bby.query.server.domain.share.util.ErrorLogUtil;
import cn.hutool.core.util.StrUtil;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author ben
 * @since 2025/4/17 00:45
 */
@Service
@Slf4j
public class InterfaceInfoManagerImpl implements InterfaceInfoManager {

    private static final String INTERFACE_INIT_LOCK_PREFIX = "interface-init-lock-";
    private static final List<Future<?>> futures = Collections.synchronizedList(new LinkedList<>());
    @Resource
    private DataStorageManager dataStorageManager;
    @Resource
    private InterfaceInfoRepository interfaceInfoRepository;
    @Resource
    private ApplicationContext applicationContext;
    @Resource(name = "dataLoadThreadPool")
    private ExecutorService dataLoadThreadPool;
    @Autowired
    private ErrorLogUtil errorLogUtil;
    @Autowired
    private RedissonClient redissonClient;

    @Value("${query.async.max-tasks:3}")
    private int maxTasks;

    @Override
    public void handleInterfaceInfos(PushInterfaceInfoReq req) {
        RLock lock = redissonClient.getLock(INTERFACE_INIT_LOCK_PREFIX + req.getServerName());
        if (!lock.tryLock()) {
            throw new ServiceException(AdvQueryErrorCode.TASK_IS_RUNNING, "该服务的接口正在初始化:" + req.getServerName());
        }
        try {
            Map<String, InterfaceInfo> infoMap = interfaceInfoRepository.getByInterfaceNames(
                            req.getServerName(),
                            req.getInterfaceMetaInfos()
                                    .stream()
                                    .map(InterfaceMetaInfo::getInterfaceName)
                                    .collect(Collectors.toList())
                    ).stream()
                    .collect(Collectors.toMap(InterfaceInfo::getMethodPath, Function.identity(), (o1, o2) -> o1));
            List<InterfaceMetaInfo> newInfos = new LinkedList<>();
            List<InterfaceMetaInfo> updateInfos = new LinkedList<>();

            for (InterfaceMetaInfo interfaceMetaInfo : req.getInterfaceMetaInfos()) {
                InterfaceInfo info = infoMap.get(interfaceMetaInfo.getInterfaceName());
                if (info != null && StrUtil.equals(info.getVersion(), interfaceMetaInfo.getVersion())) {
                    continue;
                }
                if (info != null) {
                    updateInfos.add(interfaceMetaInfo);

                } else {
                    newInfos.add(interfaceMetaInfo);
                }
            }
            doUpdateInterfaceInfos(req.getServerName(), updateInfos);
            doCreateInterfaceInfos(req.getServerName(), newInfos);
        } finally {
            lock.unlock();
        }
    }

    @Override
    @Scheduled(fixedDelay = 30, timeUnit = TimeUnit.SECONDS)
    public void loadInterfaceData() {
        if (!futures.isEmpty()) {
            futures.removeIf(Future::isDone);
        }
        // 删除失败或超时的接口数据
        List<InterfaceInfo> failList = interfaceInfoRepository.popErrorOrOutTimeList();
        if (!failList.isEmpty()) {
            dataStorageManager.deleteDataStruct(failList);
        }
        if (futures.size() >= maxTasks) {
            return;
        }
        List<InterfaceInfo> interfaceInfos = interfaceInfoRepository.popPreparedList(maxTasks - futures.size());
        if (interfaceInfos.isEmpty()) {
            return;
        }
        for (InterfaceInfo interfaceInfo : interfaceInfos) {
            futures.add(dataLoadThreadPool.submit(() -> {
                try {
                    log.info("加载接口数据: {}", interfaceInfo.getInterfaceName());
                    DataFetcher fetcher = applicationContext.getBean("KeyFullDataFetcher." + interfaceInfo.getPrimaryColumnType(), DataFetcher.class);
                    fetcher.fetchData(interfaceInfo);
                    interfaceInfoRepository.updateStatus(interfaceInfo, InterfaceStatus.READY);
                    List<InterfaceInfo> infos = interfaceInfoRepository.archiveOldNotArchiveInterfaceId(interfaceInfo);
                    dataStorageManager.deleteDataStruct(infos);
                } catch (Exception e) {
                    errorLogUtil.write(interfaceInfo.getInterfaceId(), "拉取数据失败", e);
                    interfaceInfoRepository.updateStatus(interfaceInfo, InterfaceStatus.FAILED);
                }
            }));
        }
    }

    @Override
    public List<String> reLoadInterfaceData() {
        if (futures.size() > maxTasks) {
            // 有导数任务在执行
            log.info("导数任务已满，暂不同步");
            return Collections.emptyList();
        }
        List<InterfaceInfo> interfaceInfos = interfaceInfoRepository.getReadyList(futures.size() - maxTasks);
        if (interfaceInfos.isEmpty()) {
            return Collections.emptyList();
        }
        dataLoadThreadPool.execute(() -> {
            for (InterfaceInfo interfaceInfo : interfaceInfos) {
                try {
                    log.info("重新加载接口数据: {}", interfaceInfo.getInterfaceName());
                    DataFetcher fetcher = applicationContext.getBean("KeyFullDataFetcher." + interfaceInfo.getPrimaryColumnType(), DataFetcher.class);
                    fetcher.reFetchDataAll(interfaceInfo);
                } catch (Exception e) {
                    errorLogUtil.write(interfaceInfo.getInterfaceId(), "重新拉取数据失败", e);
                }
            }
        });
        return interfaceInfos.stream().map(InterfaceInfo::getInterfaceName).collect(Collectors.toList());
    }

    private void doUpdateInterfaceInfos(String serverName, List<InterfaceMetaInfo> updateInfos) {
        if (updateInfos.isEmpty()) {
            return;
        }
        // 仅删字段

        // 加字段、改字段: 删除重建
        doCreateInterfaceInfos(serverName, updateInfos);
    }

    private void doCreateInterfaceInfos(String serverName, List<InterfaceMetaInfo> newInfos) {
        if (newInfos.isEmpty()) {
            return;
        }
        log.info("创建接口信息: {}", newInfos.size());
        interfaceInfoRepository.createBatch(serverName, newInfos);
        // 触发一次加载数据
        loadInterfaceData();
    }

}
